/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package homomorphism.list;

import homomorphism.HomomorphismBase;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;

import java.util.logging.Level;
import java.util.logging.Logger;
import utility.LongPair;
import java.io.IOException;
import java.lang.reflect.*;
import java.util.ArrayList;
import java.util.Iterator;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapred.JobConf;

/**
 *
 * @author takeshi
 */
public class FirstReducerFor3rdHomo<InVal, OtVal> extends MapReduceBase implements Reducer<LongPair, InVal, LongPair, OtVal> {

    private static final Log LOG = LogFactory.getLog(FirstReducerFor3rdHomo.class);
    private static final String userAppProperty = HomomorphismBase.UserClassNameProperty;
    private Class FoldingType;
    private Class UnFoldingType;
    private String UserClassName;
    private boolean hasEfficientMd;

    @Override
    public void configure(JobConf job) {
        super.configure(job);
        UserClassName = job.get(userAppProperty);
        FoldingType = job.getMapOutputValueClass();
        UnFoldingType = job.getOutputValueClass();
    }

    public OtVal fold(ArrayList<InVal> values) throws Exception {
        // ClassLoader myloader = FirstReducerFor3rdHomo.class.getClassLoader();
        if (UserClassName == null || UserClassName.isEmpty()) {
            throw new IOException("1 null  fold:UserClassName\n");
        }
        try {
            Class UserApp = Class.forName(UserClassName);
            //    UserApp = myloader.loadClass( UserClassName );
            Class partypes[] = new Class[1];

            ArrayList<? extends Writable> inlist = new ArrayList();
            partypes[0] = inlist.getClass();
            Method _fold = UserApp.getMethod("fold", partypes);

            Object arglist[] = new Object[1];
            arglist[0] = values;
            Constructor ct = UserApp.getConstructor(); //partypes
            Object methobj = ct.newInstance();
            OtVal rst = (OtVal) _fold.invoke(methobj, arglist);

            return rst;

        } catch (ClassNotFoundException ee) {
            LOG.error(ee.getMessage());
        }
        return null;

    }

    //new version
    public OtVal fold(Iterator<InVal> values) throws Exception {
        if (UserClassName == null || UserClassName.isEmpty()) {
            throw new IOException("1null  fold:UserClassName\n");
        }
        try {
            Class UserApp = Class.forName(UserClassName);
            //    UserApp = myloader.loadClass( UserClassName );
            Class partypes[] = new Class[1];
            partypes[0] = Iterator.class;
            Method _fold = UserApp.getMethod("fold", partypes);
            Iterator arglist[] = new Iterator[1];
            arglist[0] = values;
            Constructor ct = UserApp.getConstructor(); //partypes
            Object methobj = ct.newInstance();
            OtVal rst = (OtVal) _fold.invoke(methobj, arglist);
            return rst;

        } catch (ClassNotFoundException ee) {
            // Logger.getLogger(FirstReducerFor3rdHomo.class.getName()).log(Level.SEVERE, null, ee);
            LOG.error(ee.getMessage() + "@@@@@liu@@@@@" + values.getClass().getName());
        } catch (IllegalArgumentException ex) {
            Logger.getLogger(FirstReducerFor3rdHomo.class.getName()).log(Level.SEVERE, null, ex);
            LOG.error(ex.getMessage() + "@@@@@liu@@@@@" + values.getClass().getName() + " Iterator<InVal> is " + values.getClass().toString());
        }
        return null;

    }

    public ArrayList<InVal> unfold(OtVal value) throws Exception {
        Class UserApp = Class.forName(UserClassName);
        Class partypes[] = new Class[1];
        partypes[0] = UnFoldingType;
        Method _unfold = UserApp.getMethod("unfold", partypes);
        Object arglist[] = new Object[1];
        arglist[0] = value;
        Constructor ct = UserApp.getConstructor(); //partypes
        Object methobj = ct.newInstance();
        ArrayList<InVal> rst = (ArrayList<InVal>) _unfold.invoke(methobj, arglist);
        return rst;
    }

    public static boolean checkList(ArrayList<Object> values) {
        boolean rst = true;
        if (values.size() > 1024) {
            rst = false;
        }
        return rst;
    }

    @Override
    public void reduce(LongPair key, Iterator<InVal> values, OutputCollector<LongPair, OtVal> output, Reporter reporter)
            throws IOException {
        long pid = key.getFirst();
        LongPair outKeyPair = new LongPair(0l, pid);
        ArrayList<InVal> tmp1 = new ArrayList<InVal>();
        ArrayList<InVal> tmp2 = new ArrayList<InVal>();
        OtVal outVal = null;
        try {
            outVal = fold(values);
        } catch (Exception ex) {
            Logger.getLogger(FirstReducerFor3rdHomo.class.getName()).log(Level.SEVERE, null, ex);
        }

        /*
        while (values.hasNext()) {
        try {
        InVal pr = values.next();
        tmp1.add(pr);
        if (outVal == null) {// for the first element
        outVal = fold(tmp1);
        tmp1.clear();
        } else { // for the rest
        tmp2.clear();
        for (InVal e : unfold(outVal)) {
        tmp2.add(e);
        }
        tmp2.add(tmp1.get(0));
        outVal = fold(tmp2);
        tmp1.clear();
        }
        } catch (Exception ex) {
        Logger.getLogger(FirstReducerFor3rdHomo.class.getName()).log(Level.SEVERE, null, ex);
        }
        }
         */
        if (outVal == null) {
            throw new IOException("3null exception : outValPair\n");
        }
        output.collect(outKeyPair, outVal);
    }

}
